package com.swak.reactivex.transport;

import java.io.File;
import java.io.FileNotFoundException;
import java.io.RandomAccessFile;
import java.net.InetSocketAddress;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicBoolean;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.swak.exception.RemotingException;
import com.swak.reactivex.stream.Stream;
import com.swak.registry.URL;

import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.DefaultFileRegion;

/**
 * Netty Channel 封装.
 * 
 * @author lifeng
 */
public class NettyChannel extends Endpoint implements com.swak.reactivex.transport.Channel {

	private static final Logger logger = LoggerFactory.getLogger(NettyChannel.class);
	/**
	 * the cache for netty channel and Telnet channel
	 */
	private static final ConcurrentMap<Channel, NettyChannel> CHANNEL_MAP = new ConcurrentHashMap<>();
	/**
	 * netty channel
	 */
	private final Channel channel;

	private final AtomicBoolean active = new AtomicBoolean(false);
	private final Map<String, Object> attributes = new ConcurrentHashMap<String, Object>();

	public NettyChannel(URL url, Channel channel, ChannelHandler handler) {
		super(url, handler);
		this.channel = channel;
	}

	public Channel channel() {
		return channel;
	}

	@Override
	public void send(Object message, boolean sent) throws RemotingException {
		if (isClosed()) {
			throw new RemotingException(
					"Failed to send message " + (message == null ? "" : message.getClass().getName()) + ":" + message
							+ ", cause: Channel closed. channel: " + localAddress() + " -> " + remoteAddress());
		}

		boolean success = true;
		long timeout = 0;
		try {
			ChannelFuture future = channel.writeAndFlush(message);
			if (sent) {
				// wait timeout ms， 不会抛出异常，仅仅是返回是否执行完成
				timeout = this.getWriteTimeout();
				success = future.await(timeout);
			}

			// 这里有可能会收到 出站 pipeline 异常，但也不一定
			// 而且只有设置了 sent == true 的情形才能收到,出站pipeline 异常
			Throwable cause = future.cause();
			if (cause != null) {
				throw cause;
			}
		} catch (Throwable e) {
			removeChannelIfDisconnected(channel);
			throw new RemotingException(
					"Failed to send message " + message + " to " + remoteAddress() + ", cause: " + e.getMessage(), e);
		}
		if (!success) {
			throw new RemotingException("Failed to send message " + message + " to " + remoteAddress() + "in timeout("
					+ timeout + "ms) limit");
		}
	}

	@Override
	public void sendFile(File file, boolean sent) throws RemotingException, FileNotFoundException {
		if (isClosed()) {
			throw new RemotingException("Failed to send file " + (file == null ? "" : file.getName()) + ":" + file
					+ ", cause: Channel closed. channel: " + localAddress() + " -> " + remoteAddress());
		}

		boolean success = true;
		long timeout = 0;
		try {

			RandomAccessFile raf = null;
			int length = -1;
			try {
				raf = new RandomAccessFile(file, "r");
				length = (int) raf.length();
			} catch (Throwable t) {
				logger.error("Encode Error.", t);
				throw t;
			} finally {
				if (length < 0 && raf != null) {
					raf.close();
				}
			}

			// 发送头
			Stream stream = new Stream().setLength(length);
			channel.write(Message.forStream().setData(stream));

			// 如果是 SSL 则不支持 ZERO-COPY（这里先不处理）
			ChannelFuture future = channel.writeAndFlush(new DefaultFileRegion(raf.getChannel(), 0, length));
			if (sent) {
				// wait timeout ms
				timeout = this.getWriteTimeout();
				success = future.await(timeout);
			}
			Throwable cause = future.cause();
			if (cause != null) {
				throw cause;
			}
		} catch (Throwable e) {
			removeChannelIfDisconnected(channel);
			throw new RemotingException(
					"Failed to send file " + file + " to " + remoteAddress() + ", cause: " + e.getMessage(), e);
		}
		if (!success) {
			throw new RemotingException(
					"Failed to send file " + file + " to " + remoteAddress() + "in timeout(" + timeout + "ms) limit");
		}
	}

	@Override
	public void close() {
		try {
			super.close();
		} catch (Exception e) {
			logger.warn(e.getMessage(), e);
		}
		try {
			removeChannelIfDisconnected(channel);
		} catch (Exception e) {
			logger.warn(e.getMessage(), e);
		}
		try {
			if (logger.isInfoEnabled()) {
				logger.info("Close netty channel " + channel);
			}
			channel.close();
		} catch (Exception e) {
			logger.warn(e.getMessage(), e);
		}
	}

	@Override
	public boolean isConnected() {
		return !isClosed() && active.get();
	}

	public void markActive(boolean isActive) {
		active.set(isActive);
	}

	@Override
	public InetSocketAddress localAddress() {
		return this.channel != null ? (InetSocketAddress) this.channel.localAddress() : null;
	}

	@Override
	public InetSocketAddress remoteAddress() {
		return this.channel != null ? (InetSocketAddress) this.channel.remoteAddress() : null;
	}

	@Override
	public boolean hasAttribute(String key) {
		return this.attributes.containsKey(key);
	}

	@Override
	public Object getAttribute(String key) {
		return this.attributes.get(key);
	}

	@Override
	public void setAttribute(String key, Object value) {
		this.attributes.put(key, value);
	}

	@Override
	public void removeAttribute(String key) {
		this.attributes.remove(key);
	}

	public static NettyChannel getOrAddChannel(URL url, Channel ch, ChannelHandler handler) {
		if (ch == null) {
			return null;
		}
		NettyChannel ret = CHANNEL_MAP.get(ch);
		if (ret == null) {
			NettyChannel nettyChannel = new NettyChannel(url, ch, handler);
			if (ch.isActive()) {
				nettyChannel.markActive(true);
				ret = CHANNEL_MAP.putIfAbsent(ch, nettyChannel);
			}
			if (ret == null) {
				ret = nettyChannel;
			}
		}
		return ret;
	}

	/**
	 * Remove the inactive channel.
	 *
	 * @param ch netty channel
	 */
	public static void removeChannelIfDisconnected(Channel ch) {
		if (ch != null && !ch.isActive()) {
			NettyChannel nettyChannel = CHANNEL_MAP.remove(ch);
			if (nettyChannel != null) {
				nettyChannel.markActive(false);
			}
		}
	}

	public static void removeChannel(Channel ch) {
		if (ch != null) {
			NettyChannel nettyChannel = CHANNEL_MAP.remove(ch);
			if (nettyChannel != null) {
				nettyChannel.markActive(false);
			}
		}
	}
}
